package com.alarm.ecp;

import com.alarm.ecp.pojo.OrderEvent;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

/**
 * ProcessFunction 侧输出流示例
 */
public class OrderAlarmFlink {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 定义两个侧输出流 OutputTag
        OutputTag<OrderEvent> orderTimeoutTag = new OutputTag<OrderEvent>("orderTimeout") {};
        OutputTag<OrderEvent> highAmountTag = new OutputTag<OrderEvent>("highAmount") {};

        // 创建一个订单事件流（模拟数据）
        SingleOutputStreamOperator<OrderEvent> mainStream = env.fromElements(
                new OrderEvent("order1", "create", System.currentTimeMillis() - 70000, 5000), // 超时且高金额
                new OrderEvent("order2", "create", System.currentTimeMillis(), 12000),      // 高金额但未超时
                new OrderEvent("order3", "create", System.currentTimeMillis(), 8000),       // 正常订单
                new OrderEvent("order4", "pay", System.currentTimeMillis(), 20000),
                new OrderEvent("order1", "pay", System.currentTimeMillis() - 5000, 20000)// 已支付订单
        ).process(new ProcessFunction<OrderEvent, OrderEvent>() {

            @Override
            public void processElement(OrderEvent event, ProcessFunction<OrderEvent, OrderEvent>.Context ctx, Collector<OrderEvent> out) throws Exception {

                // 判断是否为创建但未支付的订单，并且已超时（60秒）
                if ("create".equals(event.eventType)) {
                    long currentTime = ctx.timerService().currentProcessingTime();
                    if (currentTime - event.timestamp > 60_000) {
                        ctx.output(orderTimeoutTag, event); // 第一次 Side Output：超时订单
                    }
                }

                // 判断是否为高金额订单
                if (event.amount > 10000) {
                    ctx.output(highAmountTag, event); // 第二次 Side Output：高金额订单
                }

                // 所有其他情况都发送到主输出流
                out.collect(event);
            }
        });

        // 主输出流
        mainStream.printToErr("Main Stream");

        // 获取第一个 Side Output 流：超时订单
        mainStream.getSideOutput(orderTimeoutTag).printToErr("Order Timeout Events");

        // 获取第二个 Side Output 流：高金额订单
        mainStream.getSideOutput(highAmountTag).printToErr("High Amount Orders");

        // 启动执行
        env.execute("Flink Side Output Example");
    }
}
